package com.cropdoctor.common;

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSON;
import com.cropdoctor.config.ErrorMessageConfig;
import com.cropdoctor.mapper.ErrorMapper;
import com.cropdoctor.model.constants.SearchConstants;
import com.cropdoctor.model.constants.VegeFruTableConstants;
import com.cropdoctor.model.doc.PostDoc;
import com.cropdoctor.model.doc.PublishCourseDoc;
import com.cropdoctor.model.doc.VegeFruGardenDoc;
import com.cropdoctor.model.doc.VegeFruTableDoc;
import com.cropdoctor.model.po.*;
import com.cropdoctor.util.ContextGet;


import io.redisearch.Document;
import io.redisearch.client.Client;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.security.core.parameters.P;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Component
public class EsMsgHandler {
    public static final String QUEUE_COURSE = "queue_course";

    public static final String QUEUE_POST = "queue_post";

    public static final String QUEUE_TABLE = "queue_table";

    public static final String EXCHANGE_TOPIC="exchange_topic";
    public static final String ROUTINGKEY_COURSE="COURSE";

    public static final String ROUTINGKEY_POST="POST";

    public static final String ROUTINGKEY_TABLE="TABLE";

    public static final String ERROR_DIRECT = "error.direct";
    public static final String ERROR_QUEUE = "error.queue";
    public static final String ERROR_ROUTING_KEY = "error";


    @Autowired
    ErrorMapper errorMapper;

    @Autowired
    @Qualifier(SearchConstants.TABLE_REDIS_SEARCH)
    Client tableSearchClient;
    @Qualifier(SearchConstants.POST_REDIS_SEARCH)
    @Autowired
    private Client postSearchClient;
    @Qualifier(SearchConstants.COURSE_REDIS_SEARCH)
    @Autowired
    private Client courseSearchClient;

    //当专家上传课程的时候 需要将其加入到es索引库中 以便用户可以搜索到该课程
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = QUEUE_COURSE,
                    durable = "true",
                    arguments = @Argument(name = "x-queue-mode", value = "lazy")), //声明为懒队列 保证mq消息可靠性
            exchange = @Exchange(name = EXCHANGE_TOPIC,type = ExchangeTypes.TOPIC),
            key = {ROUTINGKEY_COURSE}
    ))
    public void listenCourse(Map<String,Object> map){
        Long id = Long.valueOf(map.get("id").toString());
        String type = map.get("type").toString();//判断此处操作需要新增还是删除索引
        String docId = SearchConstants.COURSE+":"+id;
        boolean b = false;
        if("1".equals(type)){//新增索引
            String publishCourseJson = map.get("Po").toString();
            PublishCoursePo publishCoursePo = JSON.parseObject(publishCourseJson, PublishCoursePo.class);
            PublishCourseDoc publishCourseDoc = new PublishCourseDoc(publishCoursePo);

            Map<String, Object> fields = BeanUtil.beanToMap(publishCourseDoc);

            Document document = courseSearchClient.getDocument(docId);
            if(null == document){
                b = courseSearchClient.addDocument(docId, fields);
            }else{
                b = courseSearchClient.updateDocument(docId, 1.0, fields);
            }
        }else if("0".equals(type)){ //删除索引
            b = courseSearchClient.deleteDocument(docId);
        }
        if(!b){
            throw new CustomException("同步搜索引擎失败！");
        }
    }

    //管理员更新果蔬表
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = QUEUE_TABLE,
                    durable = "true",
                    arguments = @Argument(name = "x-queue-mode", value = "lazy")), //声明为懒队列 保证mq消息可靠性
            exchange = @Exchange(name = EXCHANGE_TOPIC,type = ExchangeTypes.TOPIC),
            key = {ROUTINGKEY_TABLE}
    ))
    public void listenTable(Map<String,Object> map){
        Long id = Long.valueOf(map.get("id").toString());
        String type = map.get("type").toString();//判断此处操作需要新增还是删除索引
        String docId = SearchConstants.TABLE + ":" + id;
        boolean b = false;
        if("1".equals(type)){//新增/更新 索引
            String vegeFrutablePoJson = map.get("Po").toString();
            VegeFrutablePo vegeFrutablePo = JSON.parseObject(vegeFrutablePoJson, VegeFrutablePo.class);

            Map<String, Object> fields = BeanUtil.beanToMap(vegeFrutablePo);

            Document document = tableSearchClient.getDocument(docId);
            if(null == document){
                b = tableSearchClient.addDocument(docId, fields);
            }else{
                b = tableSearchClient.updateDocument(docId, 1.0, fields);
            }
        }else if("0".equals(type)){ //删除索引
            b = tableSearchClient.deleteDocument(docId);
        }
        if(!b){
            throw new CustomException("同步搜索引擎失败！");
        }
    }

    //用户操作帖子更新索引库
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = QUEUE_POST,
                    durable = "true",
                    arguments = @Argument(name = "x-queue-mode", value = "lazy")), //声明为懒队列 保证mq消息可靠性
            exchange = @Exchange(name = EXCHANGE_TOPIC,type = ExchangeTypes.TOPIC),
            key = {ROUTINGKEY_POST}
    ))
    public void listenPost(Map<String,Object> map){
        Long id = Long.valueOf(map.get("id").toString());
        String type = map.get("type").toString();//判断此处操作需要新增还是删除索引

        boolean b = false;
        String docId = SearchConstants.POST + ":" + id;
        if("1".equals(type)){//新增索引
            String postJson = map.get("Po").toString();
            PostPo postPo = JSON.parseObject(postJson, PostPo.class);
            PostDoc postDoc = new PostDoc(postPo);

            Map<String, Object> fields = BeanUtil.beanToMap(postDoc);

            b = postSearchClient.addDocument(docId, fields);
        }else if("0".equals(type)){ //删除索引
            b = postSearchClient.deleteDocument(docId);
        }
        if(!b){
            throw new CustomException("同步搜索引擎失败!");
        }
    }


    //本地重试三次之后消息会被mq丢弃
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = ERROR_QUEUE,
                    durable = "true",
                    arguments = @Argument(name = "x-queue-mode", value = "lazy")), //声明为懒队列 保证mq消息可靠性
            exchange = @Exchange(name = ERROR_DIRECT,type = ExchangeTypes.DIRECT),
            key = {ERROR_ROUTING_KEY}
    ))
    public void listenError(Map<String,Object> map) {
        String errorId = map.get("id").toString();
        String poJson = map.get("Po") != null ? map.get("Po").toString() : "";
        String type = map.get("type") != null ? map.get("type").toString() : "";
        String sqlTable = map.get("sqlTable") != null ? map.get("sqlTable").toString() : "";
        ErrorPo errorPo = ErrorPo.builder().sqlTable(sqlTable).errorId(Long.valueOf(errorId)).type(Long.valueOf(type)).errorJson(poJson).build();

        errorMapper.insert(errorPo);
    }
}
